package com.sq.es.templete;



import com.sq.es.config.ElasticSearchConfig;
import com.sq.es.exception.ElasticSearchConnectionException;
import com.sq.es.interfaces.ElasticSearchConnectionPoll;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.util.Queue;
import java.util.concurrent.LinkedBlockingDeque;


/**
 * @Author Li sir
 * @Date 2024/1/4
 * es连接相关的操作
 */
@Component
public class ElasticSearchConnectionTemplete implements ElasticSearchConnectionPoll, ApplicationRunner {

    @Autowired
    private ElasticSearchConfig elasticSearchConfig;

    /**
     * 活跃数量 正在使用的数量
     */
    private Queue<RestHighLevelClient> activePoll;

    /**
     * 非活跃数量 未使用的数量
     */
    private Queue<RestHighLevelClient> unactivePoll;

    private static final Object lock = new Object();




    public RestHighLevelClient createElasticClient(){
        String connnectUrl = "http://"+elasticSearchConfig.getUrl()+":"+elasticSearchConfig.getPort();
        HttpHost httpHost = HttpHost.create(connnectUrl);
        RestClientBuilder restClientBuilder = RestClient.builder(httpHost);
        RestHighLevelClient restHighLevelClient = new RestHighLevelClient(restClientBuilder);
        return restHighLevelClient;
    }


    public RestHighLevelClient createElasticClient(String url,Integer port){
        String connnectUrl = "http://"+url+":"+port;
        HttpHost httpHost = HttpHost.create(connnectUrl);
        RestClientBuilder restClientBuilder = RestClient.builder(httpHost);
        RestHighLevelClient restHighLevelClient = new RestHighLevelClient(restClientBuilder);
        return restHighLevelClient;
    }



    @Override
    public void initPoll(int connectionNum) {

        activePoll = new LinkedBlockingDeque<>();
        unactivePoll = new LinkedBlockingDeque<>();

        for (int i = 0; i < connectionNum; i++){
            RestHighLevelClient elasticClient = createElasticClient();
            unactivePoll.add(elasticClient);
        }
    }

    @Override
    public RestHighLevelClient getConnection() {

        RestHighLevelClient elasticClient = null;
        // 若非活跃里的数量大于0，从非活跃里获取,并放到活跃里
        if(unactivePoll.size() > 0){
            elasticClient = unactivePoll.poll();
            activePoll.add(elasticClient);
            return elasticClient;
        }

        // 若非活跃里的数量等于0,应该判断是不是最大连接数量,如果是，则可以继续创建
        if (activePoll.size() < elasticSearchConfig.getMaxConnectNum()){
            elasticClient = createElasticClient();
            activePoll.add(elasticClient);
            return elasticClient;
        }

        // 若非活跃里的数量等于0,且活跃里的数量已经达到最大连接数量，则等待
        synchronized (lock){
            int waitTime = elasticSearchConfig.getMinWaitTime();
            try {
                while (waitTime > elasticSearchConfig.getMaxWaitTime()){
                    if (unactivePoll.size() == 0){
                        waitTime += elasticSearchConfig.getMinWaitTime();
                        lock.wait(elasticSearchConfig.getMinWaitTime());
                    }
                }
            } catch (InterruptedException e) {
                throw new ElasticSearchConnectionException("获取连接失败");
            }
        }

        return elasticClient;
    }

    @Override
    public void replaceClient(RestHighLevelClient client) {

        // 先将client从activePoll移除
        activePoll.remove(client);
        // 然后将client添加到unactivePoll
        unactivePoll.add(client);

        synchronized (lock){
            // 最后唤醒等待的线程
            lock.notify();
        }

    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        System.out.println(">>>>>>>>>>>>>>>>>> ElasticSearch 初始化 <<<<<<<<<<<<<<<<<<<<<<<");
        initPoll(elasticSearchConfig.getDefaultConnectNum());
        System.out.println(">>>>>>>>>>>>>>>>>> ElasticSearch 初始化 完成<<<<<<<<<<<<<<<<<<<<<<<");

    }
}
